package cn.doitedu.rtdw.data_sync;

import cn.doitedu.rtdw.udfs.GeohashUdf;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class Job02_地理位置信息维表同步hbase {
    public static void main(String[] args) {


        // 创建编程环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(60000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");
        env.setParallelism(1);
        env.setRuntimeMode(RuntimeExecutionMode.BATCH);

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);

        // 建表，映射mysql中的加工好的扁平化的地理位置信息表
        tenv.executeSql("CREATE TABLE tmp_area_mysql ( \n" +
                "  lat      double \n" +
                "  ,lng      double\n" +
                "  ,province string\n" +
                "  ,city     string\n" +
                "  ,region   string\n" +
                ") with (\n" +
                "   'connector' = 'jdbc',\n" +
                "   'url' = 'jdbc:mysql://doitedu:3306/realtimedw',\n" +
                "   'table-name' = 'tmp_area',   \n" +
                "   'username' = 'root'      ,   \n" +
                "   'password' = 'root'          \n" +
                ")");

        // 建表，映射hbase中的目标表（dim_area_info）
        tenv.executeSql("create table dim_area_info_hbase(\n" +
                "   geohash    string,               \n" +
                "   f ROW<province string,city string, region string>   \n" +
                ") with (\n" +
                "'connector' = 'hbase-2.2',          \n" +
                "'table-name' = 'dim_area_info',     \n" +
                "'zookeeper.quorum' = 'doitedu:2181' \n" +
                ")");


        // 写insert ...select... 语句
        // 注册自定义函数
        tenv.createTemporaryFunction("geo", GeohashUdf.class);
        tenv.executeSql("insert into dim_area_info_hbase " +
                "select                            " +
                "  geo(lat,lng) as geohash         " +
                "  ,row(province,city,region) as f " +
                "from tmp_area_mysql");


    }
}
